/**
* Created by chenly on 12/27/23.
* Description
**/
//http
#include "httplib.h"
#include "json.hpp"
//google
#include <gflags/gflags.h>
//user headers
#include "TraderSpiImpl.h"
#include "MdSpiImpl.h"
#include "CircleBuffer.h"
#include "Msg.h"
//breakpad
#include <client/linux/handler/exception_handler.h>
#include <client/linux/handler/minidump_descriptor.h>
#include "lolog/include/Lolog.h"

using namespace lyctp;
using namespace lolog;

static volatile bool kIsRunning = true;

bool dumpCallback(const google_breakpad::MinidumpDescriptor &descriptor,
                  void *context,
                  bool succeeded) {
    LOINFO() << "Dump path:" << descriptor.path() << std::endl;
    this_thread::sleep_for(chrono::milliseconds(100));
    return succeeded;
}

/*
1、开发环境，linux，c++
2、服务端开发
    a) 通过 API 进行登录，查询全部合约信息，使用缓存存储
    b) 通过 API，订阅 IC, IF, IH, IM 品种合约行情，使用队列存储；使用缓存存储
    c) 创建写文件线程，从步骤 b 队列获取，按合约代码进行存储，数据格式 csv
    d) 创建 restful 服务，对外提供2个查询接口，分别获取品种、合约行情数据
3、客户端开发
    a）创建 c++ api，同服务端进行 restful 接口调用
    b）查询合约品种接口，获取结果
    c）查询合约行情接口，获取结果


备注：
1、注意对异常正确处理、日志记录
2、对性能进行记录、测试、优化作为加分项
3、simnow 网站仅在工作日开放

 * */
void write_file(std::ofstream &out_file,
                const CThostFtdcDepthMarketDataField *data) {

    out_file << data->InstrumentID << ","
             << data->UpdateTime << "." << data->UpdateMillisec << ","
             << data->LastPrice << ","
             << data->Volume << ","
             << data->BidPrice1 << ","
             << data->BidVolume1 << ","
             << data->AskPrice1 << ","
             << data->AskVolume1 << ","
             << data->OpenInterest << ","
             << data->Turnover << std::endl;
}

void thread_write_func(CircleBuffer *circle_buffer) {
    LyMsg msg_head;
    MsgDef *tmp_msg = (MsgDef *) new char[1024 * 5];
    map<string,std::ofstream > map_file;
    while (kIsRunning ) {
        if (msg_head.read(circle_buffer, tmp_msg, 1024 * 4)) {

            CThostFtdcDepthMarketDataField *tmp_data = (CThostFtdcDepthMarketDataField *) tmp_msg->data;

            string file_path = string(tmp_data->InstrumentID)+"_market_data.csv";
            auto find_ret = map_file.find(file_path);
            if(find_ret == map_file.end()){
                auto tmp_ret = map_file.emplace(file_path,std::ofstream(file_path, std::ios::app));
                write_file(tmp_ret.first->second,tmp_data);
            } else{
                write_file(find_ret->second,tmp_data);
            }

        } else{
            std::this_thread::sleep_for(std::chrono::milliseconds(5));
        }
    }
}

DEFINE_string(broker_id,"9999", "broker id");
DEFINE_string(invester_id,"068254", "invester id");
DEFINE_string(invester_password,"501305528", "invester_password");
DEFINE_string(trade_front_addr,"tcp://180.168.146.187:10201", "trade front address");
DEFINE_string(md_front_addr,"tcp://180.168.146.187:10211", "md front address");
DEFINE_int32(port,8080, "Server port");

int main(int argc, char *argv[]) {
    google::ParseCommandLineFlags(&argc, &argv, true);
    //crash handler
    google_breakpad::MinidumpDescriptor descriptor("./");
    google_breakpad::ExceptionHandler eh(descriptor,
                                         NULL,
                                         dumpCallback,
                                         NULL,
                                         true,
                                         -1);
    //Log init
    lolog::init_logging(argv[0],5,200<<20);
    lolog::set_log_level("Debug");

    locale::global(locale("C"));

    //Get all instruments
    LOINFO() << "Get all instruments..." << endl;
    unordered_map<string,CThostFtdcInstrumentField> all_instruments;
    get_all_instruments(FLAGS_trade_front_addr, FLAGS_broker_id, FLAGS_invester_id, FLAGS_invester_password, all_instruments);


    const char * instrument_id[] = {"IC2401", "IF2401", "IH2401", "IM2401"};
    int instrument_num = 4;

    LOINFO() << "init market..." << endl;
    unique_ptr<MdSpiImpl> md_user_spi = make_unique<MdSpiImpl>(FLAGS_broker_id, FLAGS_invester_id,
                                                               FLAGS_invester_password);
    // start write thread
    thread th(thread_write_func, md_user_spi->get_circle_buffer());

    //connect market
    md_user_spi->set_subscribe_info(const_cast<char **>(instrument_id), instrument_num);
    md_user_spi->connect(FLAGS_md_front_addr);
    LOINFO() << "init market OK!" << endl;
    //start http server
    httplib::Server svr;
    svr.Get("/AllInstruments", [&](const httplib::Request &, httplib::Response &res) {
        LOINFO() << "AllInstruments receive request." <<endl;
        nlohmann::json j;
        for(auto&ins : all_instruments){
            j.push_back(ins.first);
        }

        res.set_content(j.dump(), "text/plain");
    });
    auto& cache = md_user_spi->get_cache();
    svr.Get("/Market", [&](const httplib::Request &req, httplib::Response &res) {
        LOINFO() << "Market receive request." <<endl;
        auto find_ret = req.headers.find("InstrumentID");
        if(find_ret != req.headers.end()){
            //read md from cache
            auto ret = cache.find(find_ret->second);
            if(ret != cache.end()){
                nlohmann::json j;
                j["InstrumentID"] = ret->second.InstrumentID;
                j["UpdateTime"] = ret->second.UpdateTime;
                j["UpdateMillisec"] = ret->second.UpdateMillisec;
                j["LastPrice"] = ret->second.LastPrice;
                j["Volume"] = ret->second.Volume;
                j["BidPrice1"] = ret->second.BidPrice1;
                j["BidVolume1"] = ret->second.BidVolume1;
                j["AskPrice1"] = ret->second.AskPrice1;
                j["AskVolume1"] = ret->second.AskVolume1;
                j["OpenInterest"] = ret->second.OpenInterest;
                j["Turnover"] = ret->second.Turnover;
                res.set_content(j.dump(), "text/plain");
            } else{
                res.status = httplib::StatusCode::InternalServerError_500;
                res.set_content("InstrumentID is not found", "text/plain");
            }
        } else{
            res.status = httplib::StatusCode::InternalServerError_500;
        }
    });
    LOINFO() << "Listening on "<< FLAGS_port << endl;
    svr.listen("0.0.0.0", FLAGS_port);

    kIsRunning = false;
    th.join();
    LOINFO() << "Server is gone" << endl;
    return 0;
}

